package com.gill.touka.mq;

import com.alibaba.fastjson.JSON;
import com.gill.touka.domain.dto.AuditDTO;
import com.gill.touka.service.ShareService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.transaction.annotation.Transactional;

/**
 * 测试前提：
 * 1、只有审核为PASS的请求才需要发送异步消息到RocketMQ，测试之前需要保证审核的share，对应的审核状态为NOT_YET(修改share表)
 * 2、主流程（auditById方法）发送半消息之后，会开始执行本地事务
 * 3、本地事务执行成功后，rocketmq_transaction_log表会多出1行记录
 * 4、消费端（user-content）只能等到本地事务执行成功后才能够接收到订阅topic的消息
 * 测试流程：
 * 1、在 executeLocalTransaction方法的“return RocketMQLocalTransactionState.COMMIT;”打上断点
 * 2、debug方式启动生产者（content-center）、消费者（user-center）
 * 3、访问 localhost:8082/admin/shares/audit/1 审核id为1的share
 * 4、测试数据库的rocketmq_transaction_log表多出一行记录,share表状态变为PASS，但是生产者无法消费消息，因为本地事务的执行状态还未发到broker
 * 5、放行断点，测试生产者能够消费到消息，并在bonus_event_log表创建1行记录
 * 6、接着模拟消费端产生故障，重新准备环境，再次访问，这次不放行断点，直接jps 、kill -9 生产者进程，即broker无法收到本地事务的执行情况
 * 7、重启生产者（content-center），在checkLocalTransaction方法的“return RocketMQLocalTransactionState.COMMIT;”打断点
 * 8、过一会儿，会自动执行本地事务复查checkLocalTransaction方法，确定之前本地事务的执行情况。（broker发送的）
 * 9、放行断点，若bonus_event_log表中的积分改变了，则说明，broker中的半消息会隔一段时间发送本地事务复查的信号。
 */
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "tx-test-producer-group")
public class MyLocalTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    private ShareService shareService;

    /**
     * 执行本地事务（（RocketMQ实现分布式事务2））
     * @param msg 发送到Broker的消息
     * @param arg 发送方传递的参数
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        Integer shareId = Integer.valueOf((String) headers.get("share_id"));
        // 注意：header中存储的是字符串，需要转换成对象，才能调用auditByIdWithRocketMqLog方法
        String auditDTO = headers.get("auditDTO").toString();
        AuditDTO dto = JSON.parseObject(auditDTO, AuditDTO.class);
        try {
            shareService.auditByIdWithRocketMqLog(shareId,dto,transactionId);
            log.info("本地事务执行成功");
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception exception) {
            log.warn("本地事务执行失败",exception);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 复查本地事务状态（RocketMQ实现分布式事务3）
     * broker复查生产者本地事务的执行情况
     * @param msg 发送到Broker的消息
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        if(shareService.existTransactionId(transactionId)){
            log.info("复查：本地事务执行成功");
            return RocketMQLocalTransactionState.COMMIT;
        }else{
            log.warn("复查：本地事务执行失败");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}
